package cn.azzhu.proj

import java.util.{Calendar, UUID}

import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}

import scala.util.Random

/**
 * @author azzhu
 * @create 2020-09-25 22:02:03
 */
class SimulatedEventSource extends RichParallelSourceFunction[MarketingUserBehaviour]{
  var running = true
  val channelSet = Seq("AppStore","XiaoMiStore","HuaWeiStore")
  val behaviourTypes = Seq("Browser","Click","Install","UnInstall")

  val rand = new Random()

  override def run(ctx: SourceFunction.SourceContext[MarketingUserBehaviour]): Unit = {
    while (running) {
      val userId = UUID.randomUUID().toString
      val behaviourType = behaviourTypes(rand.nextInt(behaviourTypes.size))
      val channel = channelSet(rand.nextInt(channelSet.size))
      val ts = Calendar.getInstance().getTimeInMillis

      ctx.collect(MarketingUserBehaviour(userId,behaviourType,channel,ts))
      Thread.sleep(10)
    }
  }

  override def cancel(): Unit = {
    running = false
  }
}

case class MarketingUserBehaviour(userId:String,behaviour:String,channel:String,ts:Long)
